package com.atguigu.gmall.realtime.app.dwd.db;

import com.atguigu.gmall.realtime.app.BaseSQLApp;
import com.atguigu.gmall.realtime.common.Constant;
import com.atguigu.gmall.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * @Author lzc
 * @Date 2023/1/3 10:08
 */
public class Dwd_04_DwdTradeCancelDetail extends BaseSQLApp {
    public static void main(String[] args) {
        new Dwd_04_DwdTradeCancelDetail().init(
            3004,
            2,
            "Dwd_04_DwdTradeCancelDetail"
        );
        
    }
    
    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {
        // 1. 读取 ods_db
        readOdsDb(tEnv, "Dwd_04_DwdTradeCancelDetail");
        
        // 2. 读取 dwd 层下单事务事实表
        tEnv
            .executeSql("create table dwd_trade_order_detail(" +
                            "id string," +
                            "order_id string," +
                            "user_id string," +
                            "sku_id string," +
                            "sku_name string," +
                            "province_id string," +
                            "activity_id string," +
                            "activity_rule_id string," +
                            "coupon_id string," +
                            "date_id string," +
                            "create_time string," +
                            "source_id string," +
                            "source_type string," +
                            "source_type_name string," +
                            "sku_num string," +
                            "split_original_amount string," +
                            "split_activity_amount string," +
                            "split_coupon_amount string," +
                            "split_total_amount string," +
                            "ts bigint)" + SQLUtil.getKafkaSourceDDL(Constant.TOPIC_DWD_TRADE_ORDER_DETAIL, "Dwd_04_DwdTradeCancelDetail"));
        
        // 3. 过滤出订单表(更新)
        Table orderInfo = tEnv.sqlQuery("select " +
                                            "data['id'] id," +
                                            "data['user_id'] user_id," +
                                            "data['province_id'] province_id," +
                                            "data['operate_time'] operate_time," +
                                            "ts " +
                                            "from ods_db " +
                                            "where `database`='gmall2022' " +
                                            "and `table`='order_info' " +
                                            "and `type`='update' " +
                                            "and `old`['order_status']='1001' " +
                                            "and `data`['order_status']='1003' ");
        tEnv.createTemporaryView("order_info", orderInfo);
        
        // 3. 订单表(更新)与下单事务事实表进行 join
        tEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(30 * 60 + 5));
        Table result = tEnv.sqlQuery("select " +
                                         "od.id," +
                                         "od.order_id," +
                                         "oi.user_id," +
                                         "od.sku_id," +
                                         "od.sku_name," +
                                         "oi.province_id," +
                                         "od.activity_id," +
                                         "od.activity_rule_id," +
                                         "od.coupon_id," +
                                         "date_format(oi.operate_time, 'yyyy-MM-dd') operate_date_id," +
                                         "oi.operate_time," +
                                         "od.source_id," +
                                         "od.source_type," +
                                         "od.source_type_name," +
                                         "od.sku_num," +
                                         "od.split_original_amount," +
                                         "od.split_activity_amount," +
                                         "od.split_coupon_amount," +
                                         "od.split_total_amount," +
                                         "oi.ts " +
                                         "from order_info oi " +
                                         "join dwd_trade_order_detail od " +
                                         "on oi.id=od.order_id ");
        
        
        // 4. 写出到 kafka
        tEnv.executeSql("create table dwd_trade_cancel_detail(" +
                            "id string," +
                            "order_id string," +
                            "user_id string," +
                            "sku_id string," +
                            "sku_name string," +
                            "province_id string," +
                            "activity_id string," +
                            "activity_rule_id string," +
                            "coupon_id string," +
                            "date_id string," +
                            "cancel_time string," +
                            "source_id string," +
                            "source_type_code string," +
                            "source_type_name string," +
                            "sku_num string," +
                            "split_original_amount string," +
                            "split_activity_amount string," +
                            "split_coupon_amount string," +
                            "split_total_amount string," +
                            "ts bigint " +
                            ")" + SQLUtil.getKafkaSinkDDL(Constant.TOPIC_DWD_TRADE_CANCEL_DETAIL));
        result.executeInsert("dwd_trade_cancel_detail");
    }
}
/*
取消订单事务事实表

    order_info  update   order_status=1003
    
    order_detail  不会变化  insert
    order_detail_activity 不会变化 insert
    order_detail_coupon 不会变化 insert
    
        ttl 设置: 30min+5s
        
    order_detail  不会变化  insert
    order_detail_activity 不会变化 insert
    order_detail_coupon 不会变化 insert
        join 的结果, 可以直接从下单事务事实表中读取
        
   ttl: 30min+5s
   order_info  update   order_status=1003
    join
   dwd_trade_order_detail
   
   192.168.13.238:8080
 */